package com.kuaigou.cim.mq;

import java.net.SocketException;
import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.appleframework.jms.core.consumer.MessageConusmer;
import com.appleframework.jms.core.utils.ByteUtils;
import com.appleframework.jms.rocketmq.RocketMQPushConsumer;

public abstract class ObjectMessageConsumer extends MessageConusmer {
	
	private final static Logger logger = LoggerFactory.getLogger(ObjectMessageConsumer.class);
	
	private RocketMQPushConsumer consumer;
	
	private String topic;	

	public abstract void processMessage(Object message);
			
	@Override
	public void processMessage() {
		
	}

	protected void init() throws MQClientException, SocketException {
		logger.error("ObjectMessageConsumer init-----------------------");
		logger.error("ObjectMessageConsumer topic-----------------------" + topic);
		//String ip = IpUtil.getIp();
		//logger.info("tags:" + ip + ":" + PropertyConfigurer.getString("socket.port"));
        consumer.subscribe(topic, "*");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);  
        consumer.registerMessageListener(new MessageListenerConcurrently() {
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext Context) {
                    Message msg = list.get(0);
                    logger.error(msg.toString());
                    processMessage(ByteUtils.fromByte(msg.getBody()));
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            }
        );
        consumer.start();
	}

	public void setTopic(String topic) {
		this.topic = topic.trim().replaceAll(" ", "");
	}	
	
	public void setConsumer(RocketMQPushConsumer consumer) {
		this.consumer = consumer;
	}
	
}
